/*
 * Copyright (C) 2022  即时通讯网(52im.net) & Jack Jiang.
 * The MobileIMSDK v6.1 Project.
 * All rights reserved.
 *
 * > Github地址：https://github.com/JackJiang2011/MobileIMSDK
 * > 文档地址：  http://www.52im.net/forum-89-1.html
 * > 技术社区：  http://www.52im.net/
 * > 技术交流群：320837163 (http://www.52im.net/topic-qqgroup.html)
 * > 作者公众号：“【即时通讯技术圈】”，欢迎关注！
 * > 联系作者：  http://www.52im.net/thread-2792-1-1.html
 *
 * "即时通讯网(52im.net) - 即时通讯开发者社区!" 推荐开源工程。
 *
 * QoS4SendDaemonRoot.java at 2022-7-12 16:35:57, code by Jack Jiang.
 */
package com.nami.logic.qos;


import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.nami.gate.api.request.message.RetryMsgRequest;
import com.nami.logic.bo.send.DownModeConfig;
import com.nami.logic.bo.send.PacketTransBO;
import com.nami.logic.service.RetryForwardPacketHandler;
import com.nami.logic.service.SendFailHandler;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.*;

/**
 * TODO
 *    是否替换为guava的 Table<String, String, PacketTransBO> .  table不是线程安全
 *    防止堆积太多。重试3次的失败处理调用
 *    当用户下线。Table 删除key 批量删除留存的消息，防止用户下线还在重试
 * <p>
 *   已重试3ci需要走兜底逻辑的，发布spring事件 异步处理。 添加到guava table 包装类. 当用户下线同步删除兜底策略的Table
 *   去处理3次失败的回调， 防止堆积。 妙的很
 * <p>
 * 或者
 *    不处理。
 * <p>
 * 52im 作者的qos实现
 *
 * @jack jiang
 */
@Slf4j
public class QosSendDaemonRoot {

    private boolean DEBUG = false;

    private SendFailHandler failHandler;

    private RetryForwardPacketHandler forwardPacketHandler;

    private ConcurrentSkipListMap<String, PacketTransBO> sentMessages = new ConcurrentSkipListMap<String, PacketTransBO>();
    private ConcurrentMap<String, Long> sendMessagesTimestamp = new ConcurrentHashMap<String, Long>();
    private int CHECH_INTERVAL = 5000;
    private int MESSAGES_JUST$NOW_TIME = 2 * 1000;
    private int QOS_TRY_COUNT = 3;
    private boolean _excuting = false;

    /**
     * 如果线程内出现exception 会导致线程挂掉。。。。
     */
    private ScheduledExecutorService scheduledExecutorService = null;
    private String debugTag = "";

    public void setC2CMsgSendFailService(SendFailHandler failHandler) {
        this.failHandler = failHandler;
    }

    public void setRetryHandler(RetryForwardPacketHandler forwardHandler) {
        this.forwardPacketHandler = forwardHandler;
    }


    public QosSendDaemonRoot(int CHECH_INTERVAL
        , int MESSAGES_JUST$NOW_TIME
        , int QOS_TRY_COUNT
        , boolean DEBUG, String debugTag) {
        if (CHECH_INTERVAL > 0) {
            this.CHECH_INTERVAL = CHECH_INTERVAL;
        }
        if (MESSAGES_JUST$NOW_TIME > 0) {
            this.MESSAGES_JUST$NOW_TIME = MESSAGES_JUST$NOW_TIME;
        }
        if (QOS_TRY_COUNT >= 0) {
            this.QOS_TRY_COUNT = QOS_TRY_COUNT;
        }
        this.DEBUG = DEBUG;
        this.debugTag = debugTag;
    }

    private void doTaskOnce() {
//        log.info("【NAMI" + this.debugTag + "】 开始消费发送消息队列任务：{}", sentMessages.size());
        if (!_excuting) {
            ArrayList<PacketTransBO> lostMessages = new ArrayList<PacketTransBO>();
            _excuting = true;
            try {
                if (DEBUG && sentMessages.size() > 0) {
                    log.debug("【NAMI" + this.debugTag + "】【QoS发送方】====== 消息发送质量保证线程运行中, 当前需要处理的列表长度为" + sentMessages.size() + "...");
                }
                //** 遍历HashMap方法二（在大数据量情况下，方法二的性能要5倍优于方法一）
                Iterator<Entry<String, PacketTransBO>> entryIt = sentMessages.entrySet().iterator();
                while (entryIt.hasNext()) {
                    Entry<String, PacketTransBO> entry = entryIt.next();
                    String key = entry.getKey();
                    final PacketTransBO p = entry.getValue();

                    if (p != null) {
                        if (p.getRetryCount() >= QOS_TRY_COUNT) {
                            if (DEBUG) {
                                log.debug("【NAMI" + this.debugTag + "】【QoS发送方】指纹为" + p.getFingerprint()
                                    + "的消息包重传次数已达" + p.getRetryCount() + "(最多" + QOS_TRY_COUNT + "次)上限，将判定为丢包！");
                            }
                            lostMessages.add((PacketTransBO) p.clone());
                            remove(p.getFingerprint());
                        } else {
                            //### 2015104 Bug Fix: 解决了无线网络延较大时，刚刚发出的消息在其应答包还在途中时被错误地进行重传
                            Long sendMessageTimestamp = sendMessagesTimestamp.get(key);
                            long delta = System.currentTimeMillis() - (sendMessageTimestamp == null ? 0 : sendMessageTimestamp);
                            if (delta <= MESSAGES_JUST$NOW_TIME) {
                                if (DEBUG) {
                                    log.warn("【NAMI" + this.debugTag + "】【QoS发送方】指纹为" + key + "的包距\"刚刚\"发出才" + delta
                                        + "ms(<=" + MESSAGES_JUST$NOW_TIME + "ms将被认定是\"刚刚\"), 本次不需要重传哦.");
                                }
                            } else {
                                DownModeConfig config = new DownModeConfig();
                                config.setMode(p.getDownMsgMode());
                                config.setMqTags(p.getMqTags());
                                config.setDubboIp(p.getDubboIp());
                                config.setDubboPort(p.getDubboPort());
                                forwardPacketHandler.retryHandler(newDownMsgRequest(p), config);
                                p.increaseRetryCount();
                            }
                        }
                    } else {
                        remove(key);
                    }
                }
            } catch (Exception eee) {
                if (DEBUG) {
                    log.warn("【NAMI" + this.debugTag + "】【QoS发送方】消息发送质量保证线程运行时发生异常," + eee.getMessage(), eee);
                }
            }

            if (lostMessages != null && lostMessages.size() > 0) {
                notifyMessageLost(lostMessages);
            }
            _excuting = false;
        }
    }

    @NotNull
    private static RetryMsgRequest newDownMsgRequest(PacketTransBO p) {
        RetryMsgRequest request = new RetryMsgRequest();
        request.setAppId(p.getAppKey());
        request.setPacket(p.getPacket());
        request.setIdentityNo(p.getIdentityNo());
        request.setMessageId(p.getMessageId());
        return request;
    }

    protected void notifyMessageLost(ArrayList<PacketTransBO> lostMessages) {
        failHandler.handleFailSend(lostMessages);
    }

    public QosSendDaemonRoot startup(boolean immediately) {
        stop();

        scheduledExecutorService = new ScheduledThreadPoolExecutor(1
            , ThreadFactoryBuilder.create()
            .setNamePrefix("nami-qos-send-task")
            .setDaemon(true)
            .setUncaughtExceptionHandler(((thread, throwable) ->
                log.error("[nami-qos-send--task] ThreadPool {} got exception", thread, throwable)))
            .build());
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                doTaskOnce();
            }
        }, immediately ? 0 : CHECH_INTERVAL, CHECH_INTERVAL, TimeUnit.MILLISECONDS);

        log.info("【NAMI" + this.debugTag + "】【QoS发送方】====== 消息发送质量保证线程已成功启动");
        return this;
    }

    public void stop() {
        if (scheduledExecutorService != null) {
            try {
                scheduledExecutorService.shutdown();
            } finally {
                scheduledExecutorService = null;
            }
        }
    }

    public boolean isRunning() {
        return scheduledExecutorService != null;
    }

    public boolean exist(String fingerPrint) {
        return sentMessages.get(fingerPrint) != null;
    }

    public void put(PacketTransBO p) {
        if (p == null) {
            if (DEBUG) {
                log.warn(this.debugTag + "Invalid arg p==null.");
            }
            return;
        }
        if (p.getMessageId() == null) {
            if (DEBUG) {
                log.warn(this.debugTag + "Invalid arg p.getFp() == null.");
            }
            return;
        }

        if (!p.isQos()) {
            if (DEBUG) {
                log.warn(this.debugTag + "This protocal is not QoS pkg, ignore it!");
            }
            return;
        }

        if (sentMessages.get(p.getFingerprint()) != null) {
            if (DEBUG) {
                log.warn("【NAMI" + this.debugTag + "】【QoS发送方】指纹为" + p.getFingerprint() + "的消息已经放入了发送质量保证队列，该消息为何会重复？（生成的指纹码重复？还是重复put？）");
            }
        }

        sentMessages.put(p.getFingerprint(), p);
        sendMessagesTimestamp.put(p.getFingerprint(), System.currentTimeMillis());
        log.warn("【NAMI" + this.debugTag + "】【QoS发送方】指纹为" + p.getFingerprint() + "的消息已放入发送质量保证队列");
    }

    public void remove(final String fingerPrint) {
        try {
            // remove it
            sendMessagesTimestamp.remove(fingerPrint);
            Object result = sentMessages.remove(fingerPrint);
            if (DEBUG) {
                log.warn("【NAMI" + this.debugTag + "】【QoS发送方】指纹为" + fingerPrint + "的消息已成功从发送质量保证队列中移除(可能是收到接收方的应答也可能是达到了重传的次数上限)，重试次数="
                    + (result != null ? ((PacketTransBO) result).getRetryCount() : "none呵呵."));
            }
        } catch (Exception e) {
            if (DEBUG) {
                log.warn("【NAMI" + this.debugTag + "】【QoS发送方】remove(fingerPrint)时出错了：", e);
            }
        }
    }

    public int size() {
        return sentMessages.size();
    }

    public QosSendDaemonRoot setDebugable(boolean debugable) {
        this.DEBUG = debugable;
        return this;
    }

    public boolean isDebugable() {
        return this.DEBUG;
    }
}
